package com.udf.flink.scala.apitest.trigger


import java.time.Duration

import com.udf.flink.scala.apitest.checkpoint.Obj1
import com.udf.flink.scala.apitest.window.{MinDataReduceFunction, PrintWindElenemt}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.DeltaTrigger
import org.apache.flink.streaming.api.windowing.windows.Window

object TestDeltaTrigger extends App {
  val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  //  environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  val stream1: DataStream[String] = environment.socketTextStream("localhost", 9999)
  val stream2: DataStream[Obj1] = stream1.map(data => {
    val arr = data.split(",")
    //println(arr)
    Obj1(arr(0), arr(1), arr(2).toLong)
  }).assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withTimestampAssigner(new SerializableTimestampAssigner[Obj1] {
      override def extractTimestamp(element: Obj1, recordTimestamp: Long): Long = element.time * 1000
    }))

  // 增量 获取函数
  private val deltaFunction: DeltaFunction[Obj1] = new DeltaFunction[Obj1] {
    override def getDelta(oldDataPoint: Obj1, newDataPoint: Obj1) = {
      newDataPoint.time - oldDataPoint.time
    }
  }
  // 触发器
  private val deltaTrigger: DeltaTrigger[Obj1, Window]
  = DeltaTrigger.of[Obj1, Window](10L, deltaFunction, createTypeInformation[(Obj1)].createSerializer(environment.getConfig))

  stream2.keyBy(_.id)
    .window(TumblingEventTimeWindows.of(Time.minutes(10))) // 设置窗口时间是十分钟
    .trigger(deltaTrigger)
    .reduce(new MinDataReduceFunction)
    .print("TumblingEventTimeWindow")
  environment.execute()
}

